package com.atguigu.edu.realtime.utils;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.annotaiton.NotSink;
import com.atguigu.edu.realtime.bean.TableProcess;
import com.atguigu.edu.realtime.bean.TradeCourseOrderBean;
import com.atguigu.edu.realtime.common.Constant;
import com.atguigu.edu.realtime.sink.PhoenixSink;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.shaded.guava18.com.google.common.base.CaseFormat;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.Properties;
import java.util.stream.Collectors;


public class FlinkSinkUtil {
    
    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getPhoenixSink() {
        
        return new PhoenixSink();
        
    }
    
    public static SinkFunction<String> getKafkaSink(String topic) {
        
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", Constant.KAFKA_BROKERS);
        //transaction.max.timeout.ms broker 的参数: 15 分钟, 生成者的设置不能超过这个值
        //transaction.timeout.ms  flink 的生产者默认是 60 分钟, 这个值应该小于transaction.max.timeout.ms
        props.setProperty("transaction.timeout.ms", 15 * 60 * 1000 + "");
        
        return new FlinkKafkaProducer<String>(
            "default",
            new KafkaSerializationSchema<String>() {
                @Override
                public ProducerRecord<byte[], byte[]> serialize(String element,
                                                                @Nullable Long timestamp) {
                    return new ProducerRecord<>(topic, element.getBytes(StandardCharsets.UTF_8));
                }
            },
            props,
            FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }
    
    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getKafkaSink() {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", Constant.KAFKA_BROKERS);
        props.setProperty("transaction.timeout.ms", 15 * 60 * 1000 + "");
        
        return new FlinkKafkaProducer<Tuple2<JSONObject, TableProcess>>(
            "default",
            new KafkaSerializationSchema<Tuple2<JSONObject, TableProcess>>() {
                @Override
                public ProducerRecord<byte[], byte[]> serialize(Tuple2<JSONObject, TableProcess> element,
                                                                @Nullable Long timestamp) {
                    String topic = element.f1.getSinkTable();
                    return new ProducerRecord<>(topic, element.f0.toJSONString().getBytes(StandardCharsets.UTF_8));
                }
            },
            props,
            FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }
    
    public static <T> SinkFunction<T> getClickHouseSink(String tableName, Class<T> tClass) {
        String driver = Constant.CLICKHOUSE_DRIVER;  // ru.yandex.clickhouse.ClickHouseDriver
        String url = Constant.CLICKHOUSE_URL;
        String user = Constant.CLICKHOUSE_USER;
        String password = Constant.CLICKHOUSE_PASSWORD;
        // 前提: T 类中的属性名要和表中的字段名保持一致 (驼峰或下划线)
        // insert into user(a_b,b_cc,c,d)values(?,?,?,?)
        
        final Field[] fields = tClass.getDeclaredFields();
        
        String columnNames = Arrays.stream(fields)
            // f 这个 field 如果有 NoSink 这个注解,则返回 false
            .filter(f -> f.getAnnotation(NotSink.class) == null)  // 过滤掉不需要写入到 clickhouse 的字段
            // 把驼峰转成下划线
            .map(f -> CaseFormat.LOWER_CAMEL.to(CaseFormat.LOWER_UNDERSCORE, f.getName()))
            .collect(Collectors.joining(","));
        
        StringBuilder sql = new StringBuilder();
        sql
            .append("insert into ")
            .append(tableName)
            .append("(")
            // 字段名
            .append(columnNames)
            .append(")values(")
            // 拼接占位符
            .append(columnNames.replaceAll("[^,]+", "?"))
            .append(")");
        
        System.out.println("clickhouse插入语句: " + sql);
        
        return JdbcSink.sink(sql.toString(),
                             new JdbcStatementBuilder<T>() {
                                 @Override
                                 public void accept(PreparedStatement ps,
                                                    T t) throws SQLException {
                                     Field[] fields = tClass.getDeclaredFields();
                                     // 给每个占位符赋值
                                     try {
                                         for (int i = 0, position = 1; i < fields.length; i++) {
                                             
                                             Field field = fields[i];  // t.name
                                             if (field.getAnnotation(NotSink.class) == null) {
                                                 field.setAccessible(true);  // 允许访问私有属性
    
                                                 Object v = field.get(t);
                                                 ps.setObject(position++, v);
                                             }
                                             
                                         }
                                     } catch (IllegalAccessException e) {
                                         throw new RuntimeException(e);
                                     }
                                 }
                             },
                             new JdbcExecutionOptions.Builder()
                                 .withBatchSize(1024 * 1024)
                                 .withBatchIntervalMs(5000)
                                 .withMaxRetries(3)
                                 .build(),
                             new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                                 .withDriverName(driver)
                                 .withUrl(url)
                                 .withUsername(user)
                                 .withPassword(password)
                                 .build()
        );
    }

    public static <T> SinkFunction<T> getClickHourseSink(String tableName,Class<T> tClass) {
        String driver = Constant.CLICKHOUSE_DRIVER;
        String url = Constant.CLICKHOUSE_URL;
        String user = Constant.CLICKHOUSE_USER;
        String password = Constant.CLICKHOUSE_PASSWORD;

        //获取所有属性
        Field[] fields = tClass.getDeclaredFields();
        String columnNames = Arrays
                .stream(fields)
                //filed这个属性没有NotSink这个注解，则返回false
                //过滤掉不需要写入到clickhouse的字段
                .filter(field -> field.getAnnotation(NotSink.class) == null)
                //调用map算子获取属性名（或者for循环遍历拿到属性名）
                .map(field -> com.google.common.base.CaseFormat.LOWER_CAMEL.to(com.google.common.base.CaseFormat.LOWER_UNDERSCORE, field.getName()))
                //用逗号拼接属性名
                .collect(Collectors.joining(","));

        StringBuilder sql = new StringBuilder();
        //insert into user(a,b,c,d)values(?,?,?,?)
        sql.append("insert into ")
                .append(tableName)
                .append("(")
                .append(columnNames)
                .append(")values(")
                .append(columnNames.replaceAll("[^,]+","?"))
                .append(")");
        System.out.println("clickhouse插入语句: " + sql);

        return JdbcSink.sink(
                sql.toString(),
                //给占位符赋值
                new JdbcStatementBuilder<T>() {
                    @Override
                    public void accept(PreparedStatement ps, T t) throws SQLException {
                        Field[] fields = tClass.getDeclaredFields();
                        try {
                            for (int i = 0, position = 1; i < fields.length; i++) {
                                Field field = fields[i];
                                if (field.getAnnotation(NotSink.class) == null) {
                                    field.setAccessible(true);
                                    Object v = field.get(t);
                                    ps.setObject(position++, v);
                                }
                            }
                        } catch (IllegalAccessException e) {
                            e.printStackTrace();
                        }
                    }
                },
                //设置执行参数
                new JdbcExecutionOptions.Builder()
                        .withBatchSize(1024*1024)
                        .withBatchIntervalMs(5000)
                        .withMaxRetries(3)
                        .build(),
                //设置连接参数
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(driver)
                        .withUrl(url)
                        .withUsername(user)
                        .withPassword(password)
                        .build()
        );
    }
}
